package com.dongdongshop.consumer;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.service.UserScoreAccountService;
import com.dongdongshop.service.UserScoreDetailService;
import com.dongdongshop.service.UserScoreUseService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @Description
 * @ClassName ScoreMQConsumer
 * @Author RC
 * @date 2020.12.27 17:58
 */
@Component
@RocketMQMessageListener(consumerGroup = "score_consumer",topic = "create_score")
public class ScoreMQConsumer implements RocketMQListener<Long>,
        RocketMQPushConsumerLifecycleListener {
    @Reference(timeout = 100000,check = false)
    private UserScoreAccountService userScoreAccountService;
    @Reference(timeout = 100000,check = false)
    private UserScoreDetailService userScoreDetailService;
    @Reference(timeout = 100000,check = false)
    private UserScoreUseService userScoreUseService;
    @Autowired
    private RedisTemplate redisTemplate;
    private Logger logger = LoggerFactory.getLogger(ScoreMQConsumer.class);

    @Override
    public void onMessage(Long goodsId) {
        //System.out.println("消息发送成功>>>>>>"+"拿到订单"+message+"开始生成订单");
        //int i = 9 / 0;
    }

    /**
     * 自定义消息消费重试策略（手动接管ACK确认机制）
     * CONSUME_SUCCESS、RECONSUME_LATER
     * 注意：保留原有消费者监听RocketMQListener
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt msg = list.get(0);
                try {
                    //解决幂等性问题 消费之前我们要判断这条消息有没有被消费成功
                    //如果成功消费后 将消息id最为key保存到redis中
                   /* Object o = redisTemplate.boundValueOps(msg.getMsgId()).get();
                    if(!ObjectUtil.isNull(o)){
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }*/
                    //String msgbody = new String(msg.getBody(), "utf-8");

                    String msgbody = new String(msg.getBody(), "utf-8");
                    Map<String,Object> map  = JSONObject.parseObject(msgbody);
                    String userId = (String)map.get("userId");
                    //增加积分 判断之前有没有增加过
                    TbUserScoreAccount user = userScoreAccountService.findInfoByUserId(userId);
                    int totalFee = Double.valueOf((String) map.get("totalFee")).intValue();
                    //增加积分账号  没增加就增加
                    if(user == null){ //没有
                        //1.2如果没有就创建一个新的
                        TbUserScoreAccount usa = new TbUserScoreAccount();
                        usa.setCreateTime(new Date());
                        usa.setExpendScore(0);
                        usa.setUserId(userId);
                        usa.setTotalScore(totalFee);
                        usa.setExpirTime(new Date());
                        userScoreAccountService.saveUsa(usa);
                    }else {
                        //1.1如果有就在原来的基础上增加
                        user.setTotalScore(user.getTotalScore() + totalFee);
                        user.setExpirTime(new Date());
                        userScoreAccountService.update(user);
                    }

                    //增加积分流水
                    userScoreDetailService.insertInfo(userId,totalFee);
                    //增加可用积分
                    userScoreUseService.insertInfo(userId,totalFee);
                    //int i = 9/0;
                }catch (Exception e) {
                    //e.printStackTrace();
                    logger.error("消息失败");
                    // 该条消息可以存储到DB或者LOG日志中，或其他处理方式，然后重试
                    //先获取消息的iD 消息的重试次数
                    int reconsumeTimes = msg.getReconsumeTimes();
                    logger.info("消息重试次数 为:" + reconsumeTimes);
                    if(reconsumeTimes >= 2 ){
                        //整合redis 将这条消息保存到死信队列中,(用redis充当死信队列) 当然也可以保存到mysql
                      logger.info("已经消费三次了 不再重试 ,将数据保存到私信队列中");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                   logger.info("消息的id 为 :" + msg.getMsgId());

                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                }
                //返回之前讲消息id保存到redis中
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
